package com.example.performance.r2dbc.compare;

import com.example.performance.r2dbc.entity.JDBCRowMapper;
import com.example.performance.r2dbc.entity.RecordDb;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @author zpMeng
 * @since 8/4/2025
 */
@Service
@Slf4j
public class StreamService {

    @Resource
    private JdbcTemplate jdbcTemplate;

    @Transactional(readOnly = true)
    public Flux<RecordDb> execute(String sql) {
        return Flux.usingWhen(
                Mono.fromCallable(() -> jdbcTemplate.queryForStream(
                        con -> {
                            con.setAutoCommit(false);
                            return con.prepareStatement(sql);
                        },
                        ps -> ps.setFetchSize(100),
                        new JDBCRowMapper())),
                stream -> Flux.fromStream(stream)
                        .onBackpressureBuffer(100)
                        .doOnCancel(() -> {
                            try (stream) {
                            } catch (Exception e) {
                                log.error("Stream close on error", e);
                            }
                        }),
                stream -> Mono.fromRunnable(() -> {
                            try {
                                stream.close();
                            } catch (Exception e) {
                                log.warn("Failed to close stream", e);
                            }
                        }
                ));
    }
}
